package my.thesis.bolts.filters;

import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class SplitLanguage  extends BaseRichBolt 
{
	private OutputCollector collector;
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		this.collector=collector;
	}

	public void execute(Tuple tuple) 
	{
		Long id = tuple.getLongByField("id");
		long millis = tuple.getLongByField("millis");
		String lang = tuple.getStringByField("lang");
		String hash = tuple.getStringByField("hash");
		Long ti = millis / (60 * 1000);
		
		if(lang.equals("en"))
		{
			collector.emit("english",new Values(id,millis,lang,hash,ti));
		}
		else
			
		{
			collector.emit("non-english",new Values(id,millis,lang,hash,ti));
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declareStream("english",new Fields("id","millis","lang","hash","timeInterval")); 
		declarer.declareStream("non-english",new Fields("id","millis","lang","hash","timeInterval")); 
	}

}
